
package hasthi.manager;

import static hasthi.common.constants.HasthiConstants.ELECTION_CANCEL_NOTICE_ACTION;
import static hasthi.common.constants.HasthiConstants.FAULTCODE_COORDINATOR_MSG_RECIVEDBY_NONCOORDINATOR;
import static hasthi.common.constants.HasthiConstants.FAULT_DETAIL_CURRENT_COORDINATOR;
import static hasthi.common.constants.HasthiConstants.FAULT_DETAIL_CURRENT_COORDINATOR_VAL_RETRY;
import static hasthi.common.constants.HasthiConstants.SYSTEM_STATUS_PREFIX;
import hasthi.actuators.ActionContext;
import hasthi.actuators.event.HasthiEvent;
import hasthi.common.DistributedLogger;
import hasthi.common.ErrorManager;
import hasthi.common.HasthiException;
import hasthi.common.RemoteProcessorException;
import hasthi.common.Utils;
import hasthi.common.XmlBeanUtils;
import hasthi.common.HasthiException.FaultCode;
import hasthi.common.constants.HasthiConstants;
import hasthi.common.constants.WSDMConstants;
import hasthi.common.constants.WsrfConstants;
import hasthi.common.constants.HasthiConstants.MngState;
import hasthi.common.constants.WSDMConstants.SystemState;
import hasthi.container.CommunicationContianier;
import hasthi.container.MessageContext;
import hasthi.container.MessageProcessor;
import hasthi.container.SimpleMessageContext;
import hasthi.container.p2p.MulticastContianier;
import hasthi.coordinator.CoordinatorContext;
import hasthi.coordinator.CoordinatorMessageProcessor;
import hasthi.coordinator.RemoteManagerContext;
import hasthi.instrument.FailureDetector;
import hasthi.manager.AssignResourcesDocument.AssignResources;
import hasthi.manager.AssignResourcesDocument.AssignResources.ResourceData;
import hasthi.manager.AssignResourcesResponseDocument.AssignResourcesResponse;
import hasthi.manager.CoordinatorHeartbeatDocument.CoordinatorHeartbeat;
import hasthi.manager.CoordinatorStepDownNoticeDocument.CoordinatorStepDownNotice;
import hasthi.manager.ElectedNoticeDocument.ElectedNotice;
import hasthi.manager.ElectedNoticeResponseDocument.ElectedNoticeResponse;
import hasthi.manager.ExecuteOperationDocument.ExecuteOperation;
import hasthi.manager.ExecuteQueryDocument.ExecuteQuery;
import hasthi.manager.ExecuteQueryDocument.ExecuteQuery.Query;
import hasthi.manager.GetCoordinatorResponseDocument.GetCoordinatorResponse;
import hasthi.manager.GetResourcesDocument.GetResources;
import hasthi.manager.GetResourcesResponseDocument.GetResourcesResponse;
import hasthi.manager.ListAssignedResourcesResponseDocument.ListAssignedResourcesResponse;
import hasthi.manager.ManagerPingResponseDocument.ManagerPingResponse;
import hasthi.manager.NewCoordinatorInvitationDocument.NewCoordinatorInvitation;
import hasthi.manager.NewCoordinatorInvitationResponseDocument.NewCoordinatorInvitationResponse;
import hasthi.manager.NominationRequestDocument.NominationRequest;
import hasthi.manager.NominationResponseDocument.NominationResponse;
import hasthi.manager.NotifyEventDocument.NotifyEvent;
import hasthi.manager.ResolveServiceDependancyByTypeDocument.ResolveServiceDependancyByType;
import hasthi.manager.monitering.VerificationTools;
import hasthi.metamodel.MixedResourceModel;
import hasthi.metamodel.ResourceUtils;
import hasthi.metamodel.resource.EditableManagedResource;
import hasthi.metamodel.resource.ManagedResource;
import hasthi.tools.ManagementRingTools;
import hasthi.wsrf.WsrfClientUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;

import javax.xml.namespace.QName;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.xmlbeans.XmlCursor;
import org.apache.xmlbeans.XmlException;
import org.apache.xmlbeans.XmlObject;

import wsdm.muws.ManagementEventDocument;
import xsul.soap12_util.Soap12Util;
import xsul.xbeans_util.XBeansUtil;

public class ManagerMessageProcessor implements MessageProcessor{
    

    

    private static Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);
    
    private ManagerContext managerContext;
    private CoordinatorMessageProcessor coordinatorInstance;
    private DistributedLogger dlogger;
   

    public ManagerMessageProcessor(ManagerContext managerContext) throws HasthiException {
        this.managerContext = managerContext;
        this.dlogger= managerContext.getBaseContext().getDlogger();
        managerContext.getManagementDemon().startDeamon();
//        Timer timer = new Timer(true);
//        int delay = 5000; // delay for 5 sec.
//        int period = managerContext.getBaseContext().getManagerEpochSeconds()*1000;
//        timer.scheduleAtFixedRate(managerContext.getManagementDemon(), delay, period);
    }

    public void start() throws HasthiException{
    }

    public XmlObject processMessage(MessageContext messageContext) throws HasthiException{
//        System.out.println(request.getClass());
//        System.out.println(XmlBeanUtils.prettyPrint(request));
//        String requestPath = messageContext.getSoapEndpoint();
//        String soapAction = messageContext.getSoapAction();
        XmlObject request = messageContext.getRequest();
        if(log.isTraceEnabled() && request != null){
            log.trace("Received " +XmlBeanUtils.prettyPrint(request));
        }
        try {
            if(request == null){
                return processRESTRequest(messageContext.getSoapEndpoint());
            }else if(request instanceof NominationRequestDocument){
                return processNominationRequest((NominationRequestDocument)request);
            }else if(request instanceof ElectedNoticeDocument){
                return processElectedNotice((ElectedNoticeDocument)request);
            }else if(request instanceof UnAssignResourceDocument){
                return unassignResource((UnAssignResourceDocument)request);
            }else if(request instanceof GetResourcesDocument){
                return getResources((GetResourcesDocument)request);
            }else if(request instanceof ListAssignedResourcesDocument){
                return listResources((ListAssignedResourcesDocument)request);
            }else if(request instanceof NewCoordinatorInvitationDocument){                
                return processCoordinatorInvitation((NewCoordinatorInvitationDocument)request);
            }else if(request instanceof GetCoordinatorDocument){                
                return getCoordinator((GetCoordinatorDocument)request);
            }else if(request instanceof ExecuteOperationDocument){                
                return executeOperation((ExecuteOperationDocument)request);
            }else if(request instanceof ManagerPingRequestDocument){                
                return processPing((ManagerPingRequestDocument)request);
            }else if(request instanceof ManagementEventDocument){  
                processManagmentEvent((ManagementEventDocument)request);
                return null;
            }else if(request instanceof CoordinatorStepDownNoticeDocument){  
                processElectionCancelNotice((CoordinatorStepDownNoticeDocument)request);
                return null;
            }else if(request instanceof CoordinatorHeartbeatDocument){  
                processCoordinatorHeartbeat((CoordinatorHeartbeatDocument)request);
                return null;
            }else if(managerContext.amICoordinator()){   
                coordinatorInstance = managerContext.getCoordinatorContext().getCoordinatorMessageProcessor();
                if(request instanceof JoinManagementRingDocument){
                    return coordinatorInstance.joinManagmentRing((JoinManagementRingDocument)request);
                }else if(request instanceof LeaveManagementRingDocument){
                    return coordinatorInstance.leaveManagmentRing((LeaveManagementRingDocument)request);
                }else if(request instanceof GetResourceHoldersDocument){
                    return coordinatorInstance.getResourceHolders((GetResourceHoldersDocument)request);
                }else if(request instanceof ManagerHeartbeatDocument){                
                    return coordinatorInstance.processManagerHeartBeat((ManagerHeartbeatDocument)request);
                }else if(request instanceof ListAllResourcesDocument){
                    return coordinatorInstance.listAllResources((ListAllResourcesDocument)request);
                }else if(request instanceof CoordinatorStateRequestDocument){
                    return coordinatorInstance.getCoordinatorState((CoordinatorStateRequestDocument)request);
                }else if(request instanceof UserInterventionDocument){ 
                    coordinatorInstance.processUserIntervention((UserInterventionDocument)request);
                    return null;
                }else if(request instanceof NotifyEventDocument){ 
                    coordinatorInstance.processNotifyEventDocument((NotifyEventDocument)request);
                    return null;
                }else if(request instanceof ManageMeRequestDocument 
                        || request instanceof AssignResourcesResponseDocument){
                    managerContext.getScheduler().submit(coordinatorInstance.createAsyncCoordinatorProcessor(messageContext));
                    return null;
                }else if(request instanceof ResolveServiceDependancyByTypeDocument){
                    return coordinatorInstance.resolveServiceDependancy((ResolveServiceDependancyByTypeDocument)request);
                }else if(request instanceof ExecuteQueryDocument){
                    return coordinatorInstance.processExecuateQuery((ExecuteQueryDocument)request);
                }else if(request instanceof SuspectFaultyDocument){
                    coordinatorInstance.processSuspectFaulty((SuspectFaultyDocument)request);
                    return null;
                }else if(request instanceof AssignResourcesDocument){ 
                    managerContext.getScheduler().submit(new AsyncManagerProcessor(messageContext));
                    return null;
                }
            }else if(request instanceof JoinManagementRingDocument || request instanceof LeaveManagementRingDocument 
                || request instanceof GetResourceHoldersDocument || request instanceof ManagerHeartbeatDocument){   
                //Means a Coordinator message has received from non Coordinator
                HasthiException exception = new HasthiException(FAULTCODE_COORDINATOR_MSG_RECIVEDBY_NONCOORDINATOR, 
                        FaultCode.UnsupportedMessage);
                if(managerContext.isNominted()){
                    exception.addProperty(FAULT_DETAIL_CURRENT_COORDINATOR, FAULT_DETAIL_CURRENT_COORDINATOR_VAL_RETRY);
                }else if(managerContext.getCoordinatorAddress() != null &&
                        !Utils.equals(managerContext.getCoordinatorAddress(), managerContext.getMyAddress())){
                    exception.addProperty(FAULT_DETAIL_CURRENT_COORDINATOR, managerContext.getCoordinatorAddress().getSoapAddress());
                }
                return XBeansUtil.xmlElementToXmlObject(exception.createFault(Soap12Util.getInstance()));
            }else if(request instanceof ManageMeRequestDocument || request instanceof AssignResourcesDocument){ 
                managerContext.getScheduler().submit(new AsyncManagerProcessor(messageContext));
                return null;
            }else if(request instanceof SuspectFaultyDocument){
                processSuspectFaulty((SuspectFaultyDocument)request);
                return null;
            }
            HasthiException exception = new HasthiException("Unknown Operation of type " + request.getClass() 
                    +" " +request, FaultCode.UnsupportedMessage);
            exception.printStackTrace();
            return XBeansUtil.xmlElementToXmlObject(exception.createFault(Soap12Util.getInstance()));
        } catch (HasthiException e) {
            e.printStackTrace();
            return XBeansUtil.xmlElementToXmlObject(e.createFault(Soap12Util.getInstance()));
        }
    }
    
    


    public void addContainer(CommunicationContianier container) throws HasthiException {
        //TODO override this if We need to know the containers that use this
    }
    
    
    public boolean foward2Coordinator(MessageContext messageContext) throws HasthiException{
        ManagerAddress coordinatorAddress = managerContext.getCoordinatorAddress(); 
        if(coordinatorAddress != null){
            messageContext.setSoapEndpoint(coordinatorAddress.getSoapAddress()); 
            managerContext.getUnicastContainer().send(messageContext);
            return true;
        }
        return false;
    }

    /**
     * Call for nomination
     * @param document
     * @return
     */
    
    public NominationResponseDocument processNominationRequest(NominationRequestDocument document){
        //this is only a call for nomination, the election has not started yet, We just send out our rank  
        NominationResponseDocument responseDocument = NominationResponseDocument.Factory.newInstance();
        NominationResponse nominationResponse = responseDocument.addNewNominationResponse();
        nominationResponse.setMyAddress(managerContext.getMyAddress());
        /**
         * If we do have a coordinator, and it is newer then the failed coordinator, then the election
         * starter can use that coordinator. We only send the information, and he will make the decision. 
         */
        if(managerContext.hasCoordinator()){
            NominationRequest nominationRequest = document.getNominationRequest();
            if(nominationRequest.getOldCoordinatorTimestamp() < managerContext.getCoordinatorTimeStamp()
                    && !Utils.equals(nominationRequest.getOldCoordinatorAddress(), managerContext.getCoordinatorAddress())){
                nominationResponse.setCurrentCoordinatorAddress(managerContext.getCoordinatorAddress());
            }
        }
        return responseDocument;
    }
    
    /**
     * Acked the coordinator election message
     * @param document
     * @return
     * @throws HasthiException 
     */
    public ElectedNoticeResponseDocument processElectedNotice(ElectedNoticeDocument document) throws HasthiException{
        ElectedNotice electedNotice = document.getElectedNotice();
        log.info("["+managerContext + "]Receive Elected Notice for "+electedNotice.getAddress().getSoapAddress());
       
        ElectedNoticeResponseDocument responseDocument = ElectedNoticeResponseDocument.Factory.newInstance();
        ElectedNoticeResponse response = responseDocument.addNewElectedNoticeResponse();
        
        response.setMyAddress(managerContext.getMyAddress());
        
        boolean acceptNomination = false;
        ManagerAddress alternativeAddress = null;
        long alternativeTimeStamp = -1;
        ManagerAddress electedAddress = electedNotice.getAddress();
        
        if(managerContext.amICoordinator() && electedAddress.getRank() < managerContext.getMyRank()){
            managerContext.resignCoordinator(electedAddress,electedNotice.getCoordinatorTimestamp());
        }
        
        if(managerContext.hasCoordinator() 
                && Utils.equals(managerContext.getCoordinatorAddress(), electedAddress)){
            //we laredy know about him, all is good
            acceptNomination = true;
        }else{
            //verify do we have a coordinator
            boolean isOurCoordinatorIsUp = false;
            if(managerContext.hasCoordinator()){
                if(managerContext.getCoordinatorTimeStamp() > electedNotice.getCoordinatorTimestamp()){
                    isOurCoordinatorIsUp = true;
                }else{
                    isOurCoordinatorIsUp = false;
                }
                //If our coordinator seem to be down, verify once and for all by pinging him
                if(!isOurCoordinatorIsUp){
                    isOurCoordinatorIsUp = VerificationTools.veirfyManagerISAvalible(managerContext, managerContext.getCoordinatorAddress()) != null;
                }
            }
            
            if(isOurCoordinatorIsUp && electedAddress.getRank() > managerContext.getCoordinatorAddress().getRank()){
                alternativeAddress = managerContext.getCoordinatorAddress();
                alternativeTimeStamp = managerContext.getCoordinatorTimeStamp();
                acceptNomination = false;
                log.warn("["+managerContext + "]Reject Election for "+Utils.getID(electedNotice.getAddress()) 
                        + " proposed rank= "+ electedAddress + " > my Coordinator  = "+  Utils.getID(managerContext.getCoordinatorAddress()));
            }else{
                //if I do not have a better coordinator, only concern is does he has a higher rank
                if(electedAddress.getRank() <= managerContext.getMyRank()){
                    acceptNomination = true;
                }else{
                    acceptNomination = false;
                    alternativeAddress = managerContext.getMyAddress();
                    log.warn("["+managerContext + "]Reject Election for "+Utils.getID(electedNotice.getAddress()) 
                            + " proposed = "+ electedAddress.getRank() + " > myrank = "+  alternativeAddress.getRank());
                } 
            }
        }
        
        //this it to make coordinator's manager send ack
        if(acceptNomination){
            response.setReaction(ElectedReaction.ACKED);
            managerContext.setCoordinatorAddress(electedNotice.getAddress(),electedNotice.getCoordinatorTimestamp());
        }else{
            response.setReaction(ElectedReaction.BETTER_OPTION_KNOWN);
            response.setAlternativeAddress(alternativeAddress);
            if(alternativeTimeStamp > 0){
                response.setAlternativeCoordinatorTimestamp(alternativeTimeStamp);
            }
        }
        return responseDocument;
    }
    
    public void processElectionCancelNotice(CoordinatorStepDownNoticeDocument coordinatorStepDownDocument) throws HasthiException{
        CoordinatorStepDownNotice coordinatorStepDown = coordinatorStepDownDocument.getCoordinatorStepDownNotice();
        if(Utils.equals(coordinatorStepDown.getSteppedDownCoordinator(),managerContext.getCoordinatorAddress())){
            ManagerAddress coordinatorAddress = coordinatorStepDown.getAppointedCoordinator();
            if(coordinatorAddress != null 
                    && coordinatorStepDown.getAppointedCoordinatorTimestamp() > 0 
                    && coordinatorAddress.getRank() <= managerContext.getMyRank()){
                managerContext.setCoordinatorAddress(coordinatorAddress, coordinatorStepDown.getAppointedCoordinatorTimestamp());    
            }
        }
    }
    
    public void processCoordinatorHeartbeat(CoordinatorHeartbeatDocument coordinatorHeartbeatDocument) throws HasthiException{
        try {
            CoordinatorHeartbeat coordinatorHeartbeat = coordinatorHeartbeatDocument.getCoordinatorHeartbeat();
            if(managerContext.amICoordinator()){
                if(coordinatorHeartbeat.getAddress().getRank() < managerContext.getMyRank()){
                    //there is a better coordinator out there
                    CoordinatorStepDownNoticeDocument coordinatorStepDownNoticeDocument = CoordinatorStepDownNoticeDocument.Factory.newInstance();
                    CoordinatorStepDownNotice coordinatorStepDownNotice = coordinatorStepDownNoticeDocument.addNewCoordinatorStepDownNotice();
                    coordinatorStepDownNotice.setSteppedDownCoordinator(managerContext.getMyAddress());
                    if(managerContext.getCoordinatorAddress() != null){
                        coordinatorStepDownNotice.setAppointedCoordinator(coordinatorHeartbeat.getAddress());
                        coordinatorStepDownNotice.setAppointedCoordinatorTimestamp(coordinatorHeartbeat.getCoordinatorTimestamp());
                    }
                    managerContext.resignCoordinator(coordinatorHeartbeat.getAddress(), 
                            coordinatorHeartbeat.getCoordinatorTimestamp());
                    MulticastContianier contianier = managerContext.getMulticastContianer();
                    contianier.mulitiCastAndWait(coordinatorStepDownNoticeDocument, ELECTION_CANCEL_NOTICE_ACTION, 0, 0);
                }else if(coordinatorHeartbeat.getAddress().getRank() > managerContext.getMyRank()){
                    //We send the other guy a heartbeat message, letting him know
                    CoordinatorHeartbeatDocument mycoordinatorHeartbeatDocument = CoordinatorHeartbeatDocument.Factory.newInstance();
                    CoordinatorHeartbeat myCoordinatorHeartbeat = coordinatorHeartbeatDocument.addNewCoordinatorHeartbeat();
                    myCoordinatorHeartbeat.setAddress(managerContext.getMyAddress());
                    myCoordinatorHeartbeat.setCoordinatorTimestamp(managerContext.getCoordinatorTimeStamp());
                    managerContext.getUnicastContainer().send(
                            new SimpleMessageContext(mycoordinatorHeartbeatDocument,coordinatorHeartbeat.getAddress().getSoapAddress(),
                                    HasthiConstants.COORDINATOR_HEARTBEAT_ACTION));
                }else{
                    log.fatal("["+managerContext + "]Bug:both coordinators have same rank "+coordinatorHeartbeat.getAddress().getRank() 
                            +" = " +managerContext.getMyRank());
                }
            }else{
                if(managerContext.getManagerState().equals(MngState.NoCoordWithElection)
                        || managerContext.getManagerState().equals(MngState.NoCoordWithoutElection)){
                    managerContext.setCoordinatorAddress(coordinatorHeartbeat.getAddress(),
                            coordinatorHeartbeat.getCoordinatorTimestamp());
                }
            }
        } catch (InterruptedException e) {
            throw new HasthiException(e,FaultCode.LocalError);
        }
    }
    /**
     * Return with current coordinator. Return null if no Coordinator is known
     */
    public GetCoordinatorResponseDocument getCoordinator(GetCoordinatorDocument document){
        GetCoordinatorResponseDocument responseDocument = GetCoordinatorResponseDocument.Factory.newInstance();
        GetCoordinatorResponse getCoordinatorResponse = responseDocument.addNewGetCoordinatorResponse();
        getCoordinatorResponse.setCoordinatorTimestamp(managerContext.getCoordinatorTimeStamp());
        if(managerContext.hasCoordinator()){
            getCoordinatorResponse.setAddress(managerContext.getCoordinatorAddress());    
        }else{
            getCoordinatorResponse.setNil();
        }
        return responseDocument;
    }
    
    
    /**
     * Assign a resource to this manager. It is invoked by coordinator. When invoked it fetach the Resource property document from the 
     * Resource, add resource metamodel to resource model and respond with summery of resource to Coordinator 
     * @param document
     * @return
     * @throws HasthiException
     */
    public void assignResource(AssignResourcesDocument document) throws HasthiException{
        AssignResources resource = document.getAssignResources();
        String mySoapEndpoint = managerContext.getMyAddress().getSoapAddress();
        
        
        ResourceData[] resourceDatalist =  resource.getResourceDataArray();
        AssignResourcesResponseDocument responseDocument = AssignResourcesResponseDocument.Factory.newInstance();
        AssignResourcesResponse assignResourcesResponse = responseDocument.addNewAssignResourcesResponse();
        assignResourcesResponse.setManagerAddress(managerContext.getMyAddress());
        
        ManagedResourceType[] resourceSummeries = new ManagedResourceType[resourceDatalist.length];
        int i = 0;
        for(ResourceData resourceData:resourceDatalist){
            /*
             * First we make sure we can connect to this resource,  if we can not, 
             * we forget about the assignment and Coordinator will forget it as well
             */
            //Set the Management event Sink as this Manager, letting the resource know it is assigned
            try {
                XmlObject request = WsrfClientUtils.buildUpdateResourcePropertiesRequest(WSDMConstants.MANAGMENT_EVNET_SINK_PARAM,mySoapEndpoint );
                MessageContext requestMessage = new SimpleMessageContext(request,resourceData.getManagementEndpoint(),
                        WsrfConstants.WSA_ACTION_UPDATE_RESOURCE_PROPERTY);
                managerContext.getUnicastContainer().sendReceive(requestMessage);
            
                /**
                 * Connect to the resource and create a meta-model of the resource.
                 */
                ManagedResource managedResource = ResourceUtils.createProxy4Resource(resourceData.getManagementEndpoint(), 
                        managerContext.getUnicastContainer(), managerContext.getResourceModel());
                //mark resource as it is assigned to this manager
                managerContext.getName2mangEndpointMap().put(resourceData.getName(), resourceData.getManagementEndpoint());
                managerContext.getRuleEngine().addObject(managedResource);
                //send back a summery about resource 
    
                resourceSummeries[i] = managedResource.toXml(true);
                log.info("Assigned to " + mySoapEndpoint + "("+ resourceData.getName()+ ")"+ resourceData.getManagementEndpoint());
            } catch (Exception e) {
                dlogger.errror(new StringBuffer().append("Resource JoinFailed:Unable to connect to the service "+ resourceData.getName()) 
                        .append(" running in ").append(resourceData.getManagementEndpoint()).toString(),e);
                resourceSummeries[i] = ManagedResourceType.Factory.newInstance();
                resourceSummeries[i].setName(resourceData.getName());
                resourceSummeries[i].setManagementEndpoint(null);
            }
            i++;
            
        }
        assignResourcesResponse.setResourceSummeryArray(resourceSummeries);
        
        MessageContext request = new SimpleMessageContext(responseDocument,
                managerContext.getCoordinatorAddress().getSoapAddress(),HasthiConstants.ASSIGN_RESOURCE_RESPONSE_ACTION);
        managerContext.getUnicastContainer().send(request);
    }
    
    /**
     * Remove a resource from this manager. It is invoked by Coordinator
     * @param document
     * @return
     * @throws HasthiException
     */
    
    public UnAssignResourceResponseDocument unassignResource(UnAssignResourceDocument document) throws HasthiException{
        //We do not update the resource, resource will be updated by the new assignment
        QName resourceName = document.getUnAssignResource().getName();
        //if you change here you should change Manager deamon as well
        ManagedResource resource = managerContext.getResourceModel().removeResource(resourceName);
        managerContext.getRuleEngine().removeObject(resource);
        log.info("["+managerContext+"]UnAssigned "+resourceName);
        UnAssignResourceResponseDocument responseDocument = UnAssignResourceResponseDocument.Factory.newInstance();
        responseDocument.addNewUnAssignResourceResponse();
        return responseDocument;
    }
    
    /**
     * Get the MetaModel for given list of resources
     * @param requestDocument
     * @return
     * @throws HasthiException
     */
    public GetResourcesResponseDocument getResources(GetResourcesDocument requestDocument) throws HasthiException{
        GetResources resources = requestDocument.getGetResources();
        GetResourcesResponseDocument responseDocument = GetResourcesResponseDocument.Factory.newInstance();
        GetResourcesResponse resourcesResponse = responseDocument.addNewGetResourcesResponse();
        QName[] resourceNames = resources.getResourceNameArray();
        List<ManagedResourceType> results = new ArrayList<ManagedResourceType>();
        
        for(QName name:resourceNames){
            ManagedResource resource = managerContext.getResourceModel().getLocalResource(name);
            if(resource != null){
                results.add(resource.toXml(resources.getSummerized()));
            }
        }
        resourcesResponse.setResourceArray(results.toArray(new ManagedResourceType[0]));
        return responseDocument;
    }
    
    /**
     * List resources assigned to this manager 
     * @param document
     * @return
     */
    public ListAssignedResourcesResponseDocument listResources(ListAssignedResourcesDocument document){
        ListAssignedResourcesResponseDocument responseDocument = ListAssignedResourcesResponseDocument.Factory.newInstance();
        ListAssignedResourcesResponse listResourcesResponse = responseDocument.addNewListAssignedResourcesResponse();
        Collection<QName> resourceNames = managerContext.getResourceModel().getResources();
        for(QName resourceName:resourceNames){
            listResourcesResponse.addResourceName(resourceName);
        }
        return responseDocument;
    } 
    
    /**
     * Update the metadata model related to a given resource
     * @param eventDocument
     * @throws HasthiException
     */
    public void processManagmentEvent(ManagementEventDocument eventDocument) throws HasthiException{
        QName resourceName;
        //find resource name
        String path = "declare namespace muw='http://docs.oasis-open.org/wsdm/muws1-2.xsd'; " +
        "$this//muw:ResourceId";
        XmlObject[] pathResults = eventDocument.selectPath(path, null);
        if(pathResults.length > 0){
            String resourceID = XmlBeanUtils.getStrValue(pathResults[0]);
            resourceName = Utils.uri2QName(resourceID);
            //find and add piggyback parameters
            path = "declare namespace wsdm='http://extreme.indiana.edu/wsdm/';" +
            "$this//wsdm:PiggybackParameterList";
            pathResults = eventDocument.selectPath(path, null);
            if(pathResults.length > 0){
                XmlObject piggybackParameters = pathResults[0];
                Map<QName,Object> properties  = XmlBeanUtils.extractProperties(piggybackParameters);
                MixedResourceModel resourceModel = managerContext.getResourceModel();
                EditableManagedResource resource = resourceModel.getLocalResource(resourceName);
                if(resource != null){
                    log.debug("Heartbeat recevied "+resourceName+"="+properties);
                    for(QName name:properties.keySet()){
                        resource.setProperty(name.getLocalPart(), properties.get(name));    
                    }
                    //signal to rule engine that the resource has changed
                    resource.setLastUpdated(System.currentTimeMillis());
                    managerContext.getRuleEngine().objectChanged(resource);
                    log.debug("After Heartbeat "+resource);
                    //System.out.println("heartbeat"+new Date(System.currentTimeMillis())+" " +resource.getName());
                }else{
                    //Let the resource know he is not assigned to me 
                    String resourceManagmentEndpoint = XmlBeanUtils.findStrByXpath(WSDMConstants.MNG_EVET_SENDER_ADDRESS_XPATH, eventDocument);
                    if(resourceManagmentEndpoint != null){
                        XmlObject request = WsrfClientUtils.buildUpdateResourcePropertiesRequest(WSDMConstants.MANAGMENT_EVNET_SINK_PARAM,
                                HasthiConstants.UNSPECIFIED_EPR);
                          MessageContext requestMessage = new SimpleMessageContext(request,resourceManagmentEndpoint,
                          WsrfConstants.WSA_ACTION_UPDATE_RESOURCE_PROPERTY);
                          managerContext.getUnicastContainer().sendReceive(requestMessage);
                          dlogger.info("Receive a update from "+ resourceName + " which is not assigned to me, and resource notified");
                    }else{
                        dlogger.info("[Invalied Event] "+ XmlBeanUtils.prettyPrint(eventDocument));
                    }
                }
            }else{
                log.info("[Invalied Event]No piggyback parameters found for "+ eventDocument + " ignoring the event" );
            }
        }else{
            dlogger.info("[Invalied Event]No resourceID found in event "+ eventDocument + " Ignoring the event");
        }
        
    }
    
    public NewCoordinatorInvitationResponseDocument processCoordinatorInvitation(NewCoordinatorInvitationDocument document) throws RemoteProcessorException, HasthiException{
        dlogger.info("Receive Invitation to become Coordinator "+ managerContext);
        NewCoordinatorInvitationResponseDocument responseDocument = NewCoordinatorInvitationResponseDocument.Factory.newInstance();
        NewCoordinatorInvitationResponse newCoordinatorInvitationResponse = responseDocument.addNewNewCoordinatorInvitationResponse();
        if(managerContext.amICoordinator()){
            dlogger.info("I am already coordinator "+ managerContext);
            newCoordinatorInvitationResponse.setStatus(CoordinatorInvitaionStatus.IAM_ALREADY_COORDINATOR);
            newCoordinatorInvitationResponse.setCoordinatorAddress(managerContext.getMyAddress());
            newCoordinatorInvitationResponse.setCoordinatorTimestamp(managerContext.getCoordinatorTimeStamp());
            return responseDocument;
        }
        
        NewCoordinatorInvitation coordinatorInvitation = document.getNewCoordinatorInvitation();

        // we have a newer coordinator than he knows of let him know, and stop
        if(managerContext.hasCoordinator()){
            if(Utils.equals(managerContext.getMyAddress(),managerContext.getCoordinatorAddress())){
                log.fatal("Am I coordinator says " +managerContext.amICoordinator() + " but "
                        + Utils.getID(managerContext.getMyAddress()) + " = " +Utils.getID(managerContext.getCoordinatorAddress()));
                //we expect this never happens, but if it does we need to start a election
            }else if(coordinatorInvitation.getOldCoordinatorTimestamp() < managerContext.getCoordinatorTimeStamp()){
                dlogger.info("["+managerContext + "]Newer Coordinator "+ managerContext.getCoordinatorAddress() + " already exists");
                newCoordinatorInvitationResponse.setStatus(CoordinatorInvitaionStatus.COORDINATOR_ALREADY_EXISTS);
                newCoordinatorInvitationResponse.setCoordinatorAddress(managerContext.getCoordinatorAddress());
                newCoordinatorInvitationResponse.setCoordinatorTimestamp(managerContext.getCoordinatorTimeStamp());
                return responseDocument;
            }else{
                //OK this mean he is saying current coordinator is down, verify
                ManagerPingResponse coordinatorState = VerificationTools.veirfyManagerISAvalible(managerContext,managerContext.getCoordinatorAddress());
                if(coordinatorState != null){
                      newCoordinatorInvitationResponse.setStatus(CoordinatorInvitaionStatus.COORDINATOR_NOT_DOWN);
                      newCoordinatorInvitationResponse.setCoordinatorAddress(managerContext.getCoordinatorAddress());
                      newCoordinatorInvitationResponse.setCoordinatorTimestamp(managerContext.getCoordinatorTimeStamp());
                      dlogger.info("["+managerContext + "]Coordinator is not Down "+ managerContext);
                      return responseDocument;
                }     
            }
        } 
        //OK then coordinator is down, then we make sure all the sucessors before me are down as well, this loop 
        //usually break at first check
        managerContext.setCoordinatorDown();
        ManagerAddress[] addreses = managerContext.getSucessorList();
        for(ManagerAddress managerAddress:addreses){
            if(Utils.equals(managerAddress,managerContext.getMyAddress())){
                break;
            }else{ 
                ManagerPingResponse sucessorState = VerificationTools.veirfyManagerISAvalible(managerContext,managerAddress);
                if(sucessorState != null){
                      //Second check is to make sure rank have not changed, if it has, we discard the sucessors,
                      // as the coordinator has failed earlier and new coordinator has reassign ranks
                     if(managerAddress.getRank() == sucessorState.getManagerContext().getAddress().getRank()
                             && managerAddress.getRank() < managerContext.getMyRank()){
                        //handle false alaram, tell that sucessor to start a election
                        dlogger.info("["+managerContext + "]better sucessor exists "+ Utils.getID(managerAddress));
                        newCoordinatorInvitationResponse.setStatus(CoordinatorInvitaionStatus.BETTER_SUCESSOR_EXISTS);
                        ManagementRingTools.inviteSucessor(managerContext, managerAddress,
                                managerContext.getCoordinatorAddress(),managerContext.getCoordinatorTimeStamp());
                        return responseDocument;
                     }else{
                         log.info("["+managerContext + "]Add Sucessor list discarded");
                         managerContext.setSucessorList(new ManagerAddress[0]);
                     }
                }
            }
        }

        //We try to start a election
        boolean electionStarted = managerContext.getElectionController().startElection(coordinatorInvitation.getOldCoordinatorAddress(),
                coordinatorInvitation.getOldCoordinatorTimestamp(),true); 
        if(electionStarted){
            newCoordinatorInvitationResponse.setStatus(CoordinatorInvitaionStatus.ELECTION_STARTED);
            dlogger.info("["+managerContext + "]Starting Election ");
            return responseDocument;
        }else{
            dlogger.info("["+managerContext + "]Election already started "+ managerContext);
            newCoordinatorInvitationResponse.setStatus(CoordinatorInvitaionStatus.ELECTION_ALREADY_STARTED);
            return responseDocument;
        }
    }
    
    public ExecuteOperationResponseDocument executeOperation(ExecuteOperationDocument executeOperationDocument) throws HasthiException{
        ExecuteOperation executeOperation = executeOperationDocument.getExecuteOperation();
        OperationSpecType[] operations = executeOperation.getOperationArray();
        
        for(OperationSpecType operation:operations){
            if(operation.getOperation().getLocalPart().equals(HasthiConstants.SHUTDOWN_OPERATION_NAME)){
                managerContext.getUnicastContainer().stopContainer();
                managerContext.getMulticastContianer().stopContainer();
                managerContext.getManagementDemon().stopDeamon();
                if(managerContext.amICoordinator()){
                    managerContext.getCoordinatorContext().getCoordinationDemon().stopDeamon();
                }
                dlogger.info("Manager "+managerContext+" Killed");
            }else{
                log.warn("Unknown operation "+operation.getOperation());
            }
        }
        ExecuteOperationResponseDocument responseDocument = ExecuteOperationResponseDocument.Factory.newInstance();
        responseDocument.addNewExecuteOperationResponse();
        return responseDocument;
    }
    
    public ManagerPingResponseDocument processPing(ManagerPingRequestDocument requestDocument) throws HasthiException{
        //TODO right now we only return the internal state of the manager. We can examine the invarients of the system
        //and may be kill the manager to approximate the carsh symantics
        
        //TODO this is for perf analysis
        long start = System.currentTimeMillis();
        while(System.currentTimeMillis() - start < 10){
            System.getenv();
        }
        
        ManagerPingResponseDocument responseDocument = ManagerPingResponseDocument.Factory.newInstance();
        responseDocument.addNewManagerPingResponse().setManagerContext(managerContext.toXml());
        return responseDocument;
    }
    
    public void processSuspectFaulty(SuspectFaultyDocument document ) throws HasthiException{
        QName failedServiceName = document.getSuspectFaulty().getServiceName();
        log.info("Received faulty suspect message for "+failedServiceName);
        
        EditableManagedResource resource = (EditableManagedResource)managerContext.getResourceModel().getLocalResource(failedServiceName);
        if(resource != null){
            FailureDetector failureDetector = managerContext.getBaseContext().getFailureDetector();
            //TODO For now we call the failure detector, may be later we can do more complicated algorithm like
            //taking a overall decision
            if(!failureDetector.isHealthy(resource.getManagerAgentURL())){
                resource.setState(SystemState.CrashedState);
                log.info("Resource "+ failedServiceName + " is decided failed based on faulty suspect message");
            }
        }
    }

    
    public XmlObject processRESTRequest(String url) throws HasthiException{
        //TODO this only support GET now, but we can support post as well by look at the content type, and
        //deciding on the is it SOAP/REST
        System.out.println("REST:"+url);
        if(url.contains("hasthi/resources")){
            ListAssignedResourcesDocument resourcesDocument = ListAssignedResourcesDocument.Factory.newInstance();
            resourcesDocument.addNewListAssignedResources();
            ListAssignedResourcesResponseDocument response = listResources(resourcesDocument);
            QName[] assignedResources = response.getListAssignedResourcesResponse().getResourceNameArray();
            
            GetResourcesDocument getResourcesDocument = GetResourcesDocument.Factory.newInstance();
            GetResources getResources = getResourcesDocument.addNewGetResources();
            for(QName name:assignedResources){
                getResources.addResourceName(name);
            }
            return getResources(getResourcesDocument);
        } else {
            CoordinatorContext coordinatorContext = managerContext.getCoordinatorContext();
            if(managerContext.amICoordinator() && url.contains("hasthi/allresources")){
                coordinatorInstance = coordinatorContext.getCoordinatorMessageProcessor();
                ListAllResourcesDocument resourcesDocument = ListAllResourcesDocument.Factory.newInstance();
                resourcesDocument.addNewListAllResources();
                return coordinatorInstance.listAllResources(resourcesDocument);
            }if(managerContext.amICoordinator() && url.contains(SYSTEM_STATUS_PREFIX)){
                return coordinatorContext.getSystemStat().getSystemStatusData();
            }else{
                String[] requestData = Utils.parseURL(url);
                System.out.println("REST:"+Arrays.toString(requestData));
                if(requestData != null){
                    if(requestData[0].equals("resource")){
                        GetResourcesDocument resourcesDocument = GetResourcesDocument.Factory.newInstance();
                        resourcesDocument.addNewGetResources().addResourceName(QName.valueOf(requestData[1]));
                        return getResources(resourcesDocument);
                    }else if(requestData[0].equals(HasthiConstants.LABEL_TRIGGER_EVENT)){
                        
                        String queryStr = requestData[1];
                        String name;
                        int index = queryStr.indexOf("?"); 
                        if(index > 0){
                            name = queryStr.substring(0,index);
                            queryStr = queryStr.substring(index+1);
                        }else{
                            name = queryStr;
                            queryStr = "";
                        }
                        log.debug("Trigger Event received "+ name + " " + queryStr);
                        //We will only support user interaction at the coordiantor for now
                        String message;
                        if(coordinatorContext != null){
                            ActionContext actionContext = coordinatorContext.getRuleEngine().getActionCenter().getActionContext();
                            actionContext.getEventCorrelator().triggerEvent(new HasthiEvent(name, queryStr));
                            message = "User response Accepted";
                        }else{
                            log.warn("Trigger Event ignored becouse it occuered at non coordiantor, info[ "+ name + " " + queryStr+ "]");
                            message = "User Response Ignored";
                        }
                        XmlObject response;
                        try {
                            response = XmlObject.Factory.parse("<UserResponseProcessing xmlns=\""+HasthiConstants.MANEGER_NS+"\">"
                                    +message+"</UserResponseProcessing>");
                        } catch (XmlException e) {
                            throw new HasthiException(e,FaultCode.LocalError);
                        }
                        return response;
                    }else if(managerContext.amICoordinator() 
                            && requestData[0].equals(HasthiConstants.RESOLVE_DEPENDANCY_LABEL)){
                        ResolveServiceDependancyByTypeDocument document = ResolveServiceDependancyByTypeDocument.Factory.newInstance();
                        ResolveServiceDependancyByType resolveServiceDependancyByType = document.addNewResolveServiceDependancyByType();
                        resolveServiceDependancyByType.addNewPortTypeName().setQNameValue(QName.valueOf(requestData[1]));
                        return coordinatorInstance.resolveServiceDependancy(document);
                    }else if(managerContext.amICoordinator() 
                            && requestData[0].equals(HasthiConstants.EXECUATE_QUERY_LABEL)){
                        Matcher matcher = HasthiConstants.EXECUTE_QUERY_PATTERN.matcher(requestData[1]);
                        if(matcher.matches()){
                            ExecuteQueryDocument document = ExecuteQueryDocument.Factory.newInstance();
                            ExecuteQuery execuateQuery = document.addNewExecuteQuery();
                            Query query = execuateQuery.addNewQuery();
                            query.setQueryName(matcher.group(1));
                            query.setParameterArray(matcher.group(2).split(","));
                            return coordinatorInstance.processExecuateQuery(document);
                        }else{
                            throw new HasthiException(requestData[1] + " has a wrong format, it should be query_name(parameter1,parameter2...)",FaultCode.InvaliedRequest);
                        }
                    }
                }
            }
        }
        ManagerContextDocument document = ManagerContextDocument.Factory.newInstance();
        document.setManagerContext(managerContext.toXml());
        
        XmlCursor c = document.newCursor();
        c.toStartDoc();
        c.toFirstChild();
        c.insertProcInst("xml-stylesheet", "type=\"text/xsl\" href=\"http://www.cs.indiana.edu/~hperera/dist/system.xslt\"");
        c.dispose();

        return document;
    }
    
    public class AsyncManagerProcessor implements Runnable{
        private MessageContext messageContext;
        public AsyncManagerProcessor(MessageContext messageContext) {
             
            this.messageContext = messageContext;
            
        }
        public void run(){
            XmlObject request= null;
            try {
                request = messageContext.getRequest();
                if(request instanceof ManageMeRequestDocument){
                    if(!foward2Coordinator(messageContext)){
                        log.warn("Failed to foward "+((ManageMeRequestDocument)request).getManageMeRequest()
                                + " No Coordinator Known "); 
                    }
                }else if(request instanceof AssignResourcesDocument){
                    assignResource((AssignResourcesDocument)request);
                }else{
                    log.error("Unknown message at async processing of manager"+XmlBeanUtils.prettyPrint(request));
                }
            } catch (Throwable e) {
                log.warn("Failed to process  "+request 
                        + " "+ e.getMessage());
                ErrorManager.error(e);
            }
        }
    }
}

